package com.seckill.mq;

import com.alibaba.fastjson.JSON;
import com.seckill.common.service.RedisService;
import com.seckill.dao.ItemStockDOMapper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;


@Component
public class MQConsumer {

    private DefaultMQPushConsumer consumer;

    @Value("${mq.nameserver.addr}")
    private String nameAddr;

    @Value("${mq.topicname}")
    private String topicName;

    @Autowired
    private ItemStockDOMapper itemStockDOMapper;

    @Autowired
    private RedisService<String> stringRedis;


    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer("stock_consumer_group");
        consumer.setNamesrvAddr(nameAddr);
        consumer.subscribe(topicName, "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                //实现库存真正到数据库内扣减的逻辑
                for (Message msg : msgs) {
                    String jsonString = new String(msg.getBody());
                    System.out.println("接受减库存消息: " + jsonString);
                    // 幂等消费
                    if (!stringRedis.exists(msg.getKeys())) {
                        Map<String, Object> map = JSON.parseObject(jsonString, Map.class);
                        Integer itemId = (Integer) map.get("itemId");
                        Integer amount = (Integer) map.get("amount");
                        // 减库存
                        itemStockDOMapper.decreaseStock(itemId, amount);
                        System.out.println("减库存成功: " + jsonString);
                        // 将消息的key放入Redis
                        stringRedis.setCache(msg.getKeys(), "1", 1, TimeUnit.DAYS);
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            }
        });

        consumer.start();

    }
}
